<!DOCTYPE html>



  


<html class="theme-next muse use-motion" lang="zh-Hans">
<head>
  <meta charset="UTF-8"/>
<meta http-equiv="X-UA-Compatible" content="IE=edge" />
<meta name="viewport" content="width=device-width, initial-scale=1, maximum-scale=1"/>
<meta name="theme-color" content="#222">









<meta http-equiv="Cache-Control" content="no-transform" />
<meta http-equiv="Cache-Control" content="no-siteapp" />
















  
  
  <link href="/lib/fancybox/source/jquery.fancybox.css?v=2.1.5" rel="stylesheet" type="text/css" />







<link href="/lib/font-awesome/css/font-awesome.min.css?v=4.6.2" rel="stylesheet" type="text/css" />

<link href="/css/main.css?v=5.1.4" rel="stylesheet" type="text/css" />


  <link rel="apple-touch-icon" sizes="180x180" href="/images/apple-touch-icon-next.png?v=5.1.4">


  <link rel="icon" type="image/png" sizes="32x32" href="/images/favicon-32x32-next.png?v=5.1.4">


  <link rel="icon" type="image/png" sizes="16x16" href="/images/favicon-16x16-next.png?v=5.1.4">


  <link rel="mask-icon" href="/images/logo.svg?v=5.1.4" color="#222">





  <meta name="keywords" content="Hexo, NexT" />










<meta name="description" content="上文 《源码分析 Kafka 消息发送流程》 已经详细介绍了 KafkaProducer send 方法的流程，该方法只是将消息追加到 KafKaProducer 的缓存中，并未真正的向 broker 发送消息，本文将来探讨 Kafka 的 Sender 线程。 在 KafkaProducer 中会启动一个单独的线程，其名称为 “kafka-producer-network-thread | cl">
<meta property="og:type" content="article">
<meta property="og:title" content="KafkaProducer Sender 线程详解（含详细的执行流程图）">
<meta property="og:url" content="https://www.codingw.net/posts/80780dfa.html">
<meta property="og:site_name" content="中间件兴趣圈">
<meta property="og:description" content="上文 《源码分析 Kafka 消息发送流程》 已经详细介绍了 KafkaProducer send 方法的流程，该方法只是将消息追加到 KafKaProducer 的缓存中，并未真正的向 broker 发送消息，本文将来探讨 Kafka 的 Sender 线程。 在 KafkaProducer 中会启动一个单独的线程，其名称为 “kafka-producer-network-thread | cl">
<meta property="og:locale">
<meta property="og:image" content="https://img-blog.csdnimg.cn/20191117174420482.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ByZXN0aWdlZGluZw==,size_16,color_FFFFFF,t_70">
<meta property="og:image" content="https://img-blog.csdnimg.cn/20191117182748452.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ByZXN0aWdlZGluZw==,size_16,color_FFFFFF,t_70">
<meta property="article:published_time" content="2020-10-27T15:16:01.000Z">
<meta property="article:modified_time" content="2021-04-26T12:14:15.571Z">
<meta property="article:author" content="中间件兴趣圈">
<meta property="article:tag" content="中间件">
<meta name="twitter:card" content="summary">
<meta name="twitter:image" content="https://img-blog.csdnimg.cn/20191117174420482.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ByZXN0aWdlZGluZw==,size_16,color_FFFFFF,t_70">



<script type="text/javascript" id="hexo.configurations">
  var NexT = window.NexT || {};
  var CONFIG = {
    root: '',
    scheme: 'Muse',
    version: '5.1.4',
    sidebar: {"position":"left","display":"post","offset":12,"b2t":false,"scrollpercent":false,"onmobile":false},
    fancybox: true,
    tabs: true,
    motion: {"enable":true,"async":false,"transition":{"post_block":"fadeIn","post_header":"slideDownIn","post_body":"slideDownIn","coll_header":"slideLeftIn","sidebar":"slideUpIn"}},
    duoshuo: {
      userId: '0',
      author: '博主'
    },
    algolia: {
      applicationID: '',
      apiKey: '',
      indexName: '',
      hits: {"per_page":10},
      labels: {"input_placeholder":"Search for Posts","hits_empty":"We didn't find any results for the search: ${query}","hits_stats":"${hits} results found in ${time} ms"}
    }
  };
</script>



  <link rel="canonical" href="https://www.codingw.net/posts/80780dfa.html"/>





  <title>KafkaProducer Sender 线程详解（含详细的执行流程图） | 中间件兴趣圈</title>
  








<meta name="generator" content="Hexo 5.4.0"></head>

<body itemscope itemtype="http://schema.org/WebPage" lang="zh-Hans">

  
  
    
  

  <div class="container sidebar-position-left page-post-detail">
    <div class="headband"></div>

    <header id="header" class="header" itemscope itemtype="http://schema.org/WPHeader">
      <div class="header-inner"><div class="site-brand-wrapper">
  <div class="site-meta ">
    

    <div class="custom-logo-site-title">
      <a href="/"  class="brand" rel="start">
        <span class="logo-line-before"><i></i></span>
        <span class="site-title">中间件兴趣圈</span>
        <span class="logo-line-after"><i></i></span>
      </a>
    </div>
      
        <p class="site-subtitle">微信搜『中间件兴趣圈』，回复『Java』获取200本优质电子书</p>
      
  </div>

  <div class="site-nav-toggle">
    <button>
      <span class="btn-bar"></span>
      <span class="btn-bar"></span>
      <span class="btn-bar"></span>
    </button>
  </div>
</div>

<nav class="site-nav">
  

  
    <ul id="menu" class="menu">
      
        
        <li class="menu-item menu-item-home">
          <a href="/" rel="section">
            
              <i class="menu-item-icon fa fa-fw fa-home"></i> <br />
            
            首页
          </a>
        </li>
      
        
        <li class="menu-item menu-item-categories">
          <a href="/categories/" rel="section">
            
              <i class="menu-item-icon fa fa-fw fa-question-circle"></i> <br />
            
            分类
          </a>
        </li>
      
        
        <li class="menu-item menu-item-archives">
          <a href="/archives/" rel="section">
            
              <i class="menu-item-icon fa fa-fw fa-question-circle"></i> <br />
            
            归档
          </a>
        </li>
      

      
    </ul>
  

  
</nav>



 </div>
    </header>

    <main id="main" class="main">
      <div class="main-inner">
        <div class="content-wrap">
          <div id="content" class="content">
            

  <div id="posts" class="posts-expand">
    

  

  
  
  

  <article class="post post-type-normal" itemscope itemtype="http://schema.org/Article">
  
  
  
  <div class="post-block">
    <link itemprop="mainEntityOfPage" href="https://www.codingw.net/posts/80780dfa.html">

    <span hidden itemprop="author" itemscope itemtype="http://schema.org/Person">
      <meta itemprop="name" content="">
      <meta itemprop="description" content="">
      <meta itemprop="image" content="/images/avatar.gif">
    </span>

    <span hidden itemprop="publisher" itemscope itemtype="http://schema.org/Organization">
      <meta itemprop="name" content="中间件兴趣圈">
    </span>

    
      <header class="post-header">

        
        
          <h1 class="post-title" itemprop="name headline">KafkaProducer Sender 线程详解（含详细的执行流程图）</h1>
        

        <div class="post-meta">
          <span class="post-time">
            
              <span class="post-meta-item-icon">
                <i class="fa fa-calendar-o"></i>
              </span>
              
                <span class="post-meta-item-text">发表于</span>
              
              <time title="创建于" itemprop="dateCreated datePublished" datetime="2020-10-27T23:16:01+08:00">
                2020-10-27
              </time>
            

            

            
          </span>

          
            <span class="post-category" >
            
              <span class="post-meta-divider">|</span>
            
              <span class="post-meta-item-icon">
                <i class="fa fa-folder-o"></i>
              </span>
              
                <span class="post-meta-item-text">分类于</span>
              
              
                <span itemprop="about" itemscope itemtype="http://schema.org/Thing">
                  <a href="/categories/kafka/" itemprop="url" rel="index">
                    <span itemprop="name">kafka</span>
                  </a>
                </span>

                
                
              
            </span>
          

          
            
          

          
          
             <span id="/posts/80780dfa.html" class="leancloud_visitors" data-flag-title="KafkaProducer Sender 线程详解（含详细的执行流程图）">
               <span class="post-meta-divider">|</span>
               <span class="post-meta-item-icon">
                 <i class="fa fa-eye"></i>
               </span>
               
                 <span class="post-meta-item-text">阅读次数&#58;</span>
               
                 <span class="leancloud-visitors-count"></span>
             </span>
          

          
            <span class="post-meta-divider">|</span>
            <span class="page-pv"><i class="fa fa-file-o"></i>
            <span class="busuanzi-value" id="busuanzi_value_page_pv" ></span>次
            </span>
          

          

          

        </div>
      </header>
    

    
    
    
    <div class="post-body" itemprop="articleBody">

      
      

      
        <div id="vip-container"><p>上文 <a target="_blank" rel="noopener" href="https://blog.csdn.net/prestigeding/article/details/102994716">《源码分析 Kafka 消息发送流程》</a> 已经详细介绍了 KafkaProducer send 方法的流程，该方法只是将消息追加到 KafKaProducer 的缓存中，并未真正的向 broker 发送消息，本文将来探讨 Kafka 的 Sender 线程。</p>
<p>在 KafkaProducer 中会启动一个单独的线程，其名称为 “kafka-producer-network-thread | clientID”，其中 clientID 为生产者的 id 。</p>
<h2 id="1、Sender-线程详解"><a href="#1、Sender-线程详解" class="headerlink" title="1、Sender 线程详解"></a>1、Sender 线程详解</h2><h4 id="1-1-类图"><a href="#1-1-类图" class="headerlink" title="1.1 类图"></a>1.1 类图</h4><p><img src="https://img-blog.csdnimg.cn/20191117174420482.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ByZXN0aWdlZGluZw==,size_16,color_FFFFFF,t_70" alt="在这里插入图片描述"><br>我们先来看一下其各个属性的含义：</p>
<ul>
<li>KafkaClient client<br>kafka 网络通信客户端，主要封装与 broker 的网络通信。</li>
<li>RecordAccumulator accumulator<br>消息记录累积器，消息追加的入口(RecordAccumulator 的 append 方法)。</li>
<li>Metadata metadata<br>元数据管理器，即 topic 的路由分区信息。</li>
<li>boolean guaranteeMessageOrder<br>是否需要保证消息的顺序性。</li>
<li>int maxRequestSize<br>调用 send 方法发送的最大请求大小，包括 key、消息体序列化后的消息总大小不能超过该值。通过参数 max.request.size 来设置。</li>
<li>short acks<br>用来定义消息“已提交”的条件(标准)，就是 Broker 端向客户端承偌已提交的条件，可选值如下0、-1、1.</li>
<li>int retries<br>重试次数。</li>
<li>Time time<br>时间工具类。</li>
<li>boolean running<br>该线程状态，为 true 表示运行中。</li>
<li>boolean forceClose<br>是否强制关闭，此时会忽略正在发送中的消息。</li>
<li>SenderMetrics sensors<br>消息发送相关的统计指标收集器。</li>
<li>int requestTimeoutMs<br>请求的超时时间。</li>
<li>long retryBackoffMs<br>请求失败之在重试之前等待的时间。</li>
<li>ApiVersions apiVersions<br>API版本信息。</li>
<li>TransactionManager transactionManager<br>事务处理器。</li>
<li>Map&lt; TopicPartition, List&lt; ProducerBatch&gt;&gt; inFlightBatches<br>正在执行发送相关的消息批次。</li>
</ul>
<span id="more"></span>

<h4 id="1-2-run-方法详解"><a href="#1-2-run-方法详解" class="headerlink" title="1.2 run 方法详解"></a>1.2 run 方法详解</h4><p>Sender#run</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br></pre></td><td class="code"><pre><span class="line"><span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">run</span><span class="params">()</span> </span>&#123;</span><br><span class="line">    log.debug(<span class="string">&quot;Starting Kafka producer I/O thread.&quot;</span>);</span><br><span class="line">    <span class="keyword">while</span> (running) &#123;   </span><br><span class="line">        <span class="keyword">try</span> &#123;</span><br><span class="line">            runOnce();    <span class="comment">// @1</span></span><br><span class="line">        &#125; <span class="keyword">catch</span> (Exception e) &#123;</span><br><span class="line">            log.error(<span class="string">&quot;Uncaught error in kafka producer I/O thread: &quot;</span>, e);</span><br><span class="line">        &#125;</span><br><span class="line">    &#125;</span><br><span class="line">    log.debug(<span class="string">&quot;Beginning shutdown of Kafka producer I/O thread, sending remaining records.&quot;</span>);</span><br><span class="line">    <span class="keyword">while</span> (!forceClose &amp;&amp; (<span class="keyword">this</span>.accumulator.hasUndrained() || <span class="keyword">this</span>.client.inFlightRequestCount() &gt; <span class="number">0</span>)) &#123;    <span class="comment">// @2</span></span><br><span class="line">        <span class="keyword">try</span> &#123;</span><br><span class="line">            runOnce();</span><br><span class="line">        &#125; <span class="keyword">catch</span> (Exception e) &#123;</span><br><span class="line">            log.error(<span class="string">&quot;Uncaught error in kafka producer I/O thread: &quot;</span>, e);</span><br><span class="line">        &#125;</span><br><span class="line">    &#125;</span><br><span class="line">    <span class="keyword">if</span> (forceClose) &#123;                                                                                                                                     <span class="comment">// @3</span></span><br><span class="line">        log.debug(<span class="string">&quot;Aborting incomplete batches due to forced shutdown&quot;</span>);</span><br><span class="line">        <span class="keyword">this</span>.accumulator.abortIncompleteBatches();</span><br><span class="line">    &#125;</span><br><span class="line">    <span class="keyword">try</span> &#123;</span><br><span class="line">        <span class="keyword">this</span>.client.close();                                                                                                                               <span class="comment">// @4</span></span><br><span class="line">    &#125; <span class="keyword">catch</span> (Exception e) &#123;</span><br><span class="line">        log.error(<span class="string">&quot;Failed to close network client&quot;</span>, e);</span><br><span class="line">    &#125;</span><br><span class="line">    log.debug(<span class="string">&quot;Shutdown of Kafka producer I/O thread has completed.&quot;</span>);</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<p>代码@1：Sender 线程在运行状态下主要的业务处理方法，将消息缓存区中的消息向 broker 发送。<br>代码@2：如果主动关闭 Sender 线程，如果不是强制关闭，则如果缓存区还有消息待发送，再次调用 runOnce 方法将剩余的消息发送完毕后再退出。<br>代码@3：如果强制关闭 Sender 线程，则拒绝未完成提交的消息。<br>代码@4：关闭 Kafka Client 即网络通信对象。</p>
<p>接下来将分别探讨其上述方法的实现细节。</p>
<h5 id="1-2-1-runOnce-详解"><a href="#1-2-1-runOnce-详解" class="headerlink" title="1.2.1 runOnce 详解"></a>1.2.1 runOnce 详解</h5><p>Sender#runOnce</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br></pre></td><td class="code"><pre><span class="line"><span class="function"><span class="keyword">void</span> <span class="title">runOnce</span><span class="params">()</span> </span>&#123;</span><br><span class="line">	<span class="comment">// 此处省略与事务消息相关的逻辑</span></span><br><span class="line">    <span class="keyword">long</span> currentTimeMs = time.milliseconds();</span><br><span class="line">    <span class="keyword">long</span> pollTimeout = sendProducerData(currentTimeMs);   <span class="comment">// @1</span></span><br><span class="line">    client.poll(pollTimeout, currentTimeMs);                            <span class="comment">// @2</span></span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<p>本文不关注事务消息的实现原理，故省略了该部分的代码。<br>代码@1：调用 sendProducerData 方法发送消息。<br>代码@2：调用这个方法的作用？</p>
<p>接下来分别对上述两个方法进行深入探究。</p>
<h6 id="1-1-2-1-sendProducerData"><a href="#1-1-2-1-sendProducerData" class="headerlink" title="1.1.2.1 sendProducerData"></a>1.1.2.1 sendProducerData</h6><p>接下来将详细分析其实现步骤。<br>Sender#sendProducerData</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br></pre></td><td class="code"><pre><span class="line">Cluster cluster = metadata.fetch();</span><br><span class="line"><span class="comment">// get the list of partitions with data ready to send</span></span><br><span class="line">RecordAccumulator.ReadyCheckResult result = <span class="keyword">this</span>.accumulator.ready(cluster, now);</span><br></pre></td></tr></table></figure>
<p>Step1：首先根据当前时间，根据缓存队列中的数据判断哪些 topic 的 哪些分区已经达到发送条件。达到可发送的条件将在 2.1.1.1 节详细分析。</p>
<p>Sender#sendProducerData</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">if</span> (!result.unknownLeaderTopics.isEmpty()) &#123;</span><br><span class="line">    <span class="keyword">for</span> (String topic : result.unknownLeaderTopics)</span><br><span class="line">        <span class="keyword">this</span>.metadata.add(topic);</span><br><span class="line">    </span><br><span class="line">    log.debug(<span class="string">&quot;Requesting metadata update due to unknown leader topics from the batched records: &#123;&#125;&quot;</span>,</span><br><span class="line">                result.unknownLeaderTopics);</span><br><span class="line">    <span class="keyword">this</span>.metadata.requestUpdate();</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<p>Step2：如果在待发送的消息未找到其路由信息，则需要首先去 broker 服务器拉取对应的路由信息(分区的 leader 节点信息)。</p>
<p>Sender#sendProducerData</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">long</span> notReadyTimeout = Long.MAX_VALUE;</span><br><span class="line"><span class="keyword">while</span> (iter.hasNext()) &#123;</span><br><span class="line">    Node node = iter.next();</span><br><span class="line">    <span class="keyword">if</span> (!<span class="keyword">this</span>.client.ready(node, now)) &#123;</span><br><span class="line">        iter.remove();</span><br><span class="line">        notReadyTimeout = Math.min(notReadyTimeout, <span class="keyword">this</span>.client.pollDelayMs(node, now));</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<p>Step3：移除在网络层面没有准备好的分区，并且计算在接下来多久的时间间隔内，该分区都将处于未准备状态。<br>1、在网络环节没有准备好的标准如下：</p>
<ul>
<li>分区没有未完成的更新元素数据请求(metadata)。</li>
<li>当前生产者与对端 broker 已建立连接并完成了 TCP 的三次握手。</li>
<li>如果启用 SSL、ACL 等机制，相关状态都已就绪。</li>
<li>该分区对应的连接正在处理中的请求数时是否超过设定值，默认为 5，可通过属性 max.in.flight.requests.per.connection 来设置。</li>
</ul>
<p>2、client pollDelayMs 预估分区在接下来多久的时间间隔内都将处于未转变好状态(not ready)，其标准如下：</p>
<ul>
<li>如果已与对端的 TCP 连接已创建好，并处于已连接状态，此时如果没有触发限流，则返回0，如果有触发限流，则返回限流等待时间。</li>
<li>如果还位于对端建立 TCP 连接，则返回 Long.MAX_VALUE，因为连接建立好后，会唤醒发送线程的。</li>
</ul>
<p>Sender#sendProducerData</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">// create produce requests</span></span><br><span class="line">Map&lt;Integer, List&lt;ProducerBatch&gt;&gt; batches = <span class="keyword">this</span>.accumulator.drain(cluster, result.readyNodes, <span class="keyword">this</span>.maxRequestSize, now);</span><br></pre></td></tr></table></figure>
<p>Step4：根据已准备的分区，从缓存区中抽取待发送的消息批次(ProducerBatch)，并且按照 nodeId:List<ProducerBatch> 组织，注意，抽取后的 ProducerBatch 将不能再追加消息了，就算还有剩余空间可用，具体抽取将在下文在详细介绍。</p>
<p>Sender#sendProducerData</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br></pre></td><td class="code"><pre><span class="line">addToInflightBatches(batches);</span><br><span class="line"><span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">addToInflightBatches</span><span class="params">(Map&lt;Integer, List&lt;ProducerBatch&gt;&gt; batches)</span> </span>&#123;</span><br><span class="line">    <span class="keyword">for</span> (List&lt;ProducerBatch&gt; batchList : batches.values()) &#123;</span><br><span class="line">        addToInflightBatches(batchList);</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br><span class="line"><span class="function"><span class="keyword">private</span> <span class="keyword">void</span> <span class="title">addToInflightBatches</span><span class="params">(List&lt;ProducerBatch&gt; batches)</span> </span>&#123;</span><br><span class="line">    <span class="keyword">for</span> (ProducerBatch batch : batches) &#123;</span><br><span class="line">        List&lt;ProducerBatch&gt; inflightBatchList = inFlightBatches.get(batch.topicPartition);</span><br><span class="line">        <span class="keyword">if</span> (inflightBatchList == <span class="keyword">null</span>) &#123;</span><br><span class="line">            inflightBatchList = <span class="keyword">new</span> ArrayList&lt;&gt;();</span><br><span class="line">            inFlightBatches.put(batch.topicPartition, inflightBatchList);</span><br><span class="line">        &#125;</span><br><span class="line">        inflightBatchList.add(batch);</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<p>Step5：将抽取的 ProducerBatch 加入到 inFlightBatches 数据结构，该属性的声明如下：Map&lt;TopicPartition, List&lt; ProducerBatch &gt;&gt; inFlightBatches，即按照 topic-分区 为键，存放已抽取的 ProducerBatch，这个属性的含义就是存储待发送的消息批次。可以根据该数据结构得知在消息发送时以分区为维度反馈 Sender 线程的“积压情况”，max.in.flight.requests.per.connection 就是来控制积压的最大数量，如果积压达到这个数值，针对该队列的消息发送会限流。</p>
<p>Sender#sendProducerData</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br></pre></td><td class="code"><pre><span class="line">accumulator.resetNextBatchExpiryTime();</span><br><span class="line">List&lt;ProducerBatch&gt; expiredInflightBatches = getExpiredInflightBatches(now);</span><br><span class="line">List&lt;ProducerBatch&gt; expiredBatches = <span class="keyword">this</span>.accumulator.expiredBatches(now);</span><br><span class="line">expiredBatches.addAll(expiredInflightBatches);</span><br></pre></td></tr></table></figure>
<p>Step6：从 inflightBatches 与 batches 中查找已过期的消息批次(ProducerBatch)，判断是否过期的标准是系统当前时间与 ProducerBatch 创建时间之差是否超过120s，过期时间可以通过参数 delivery.timeout.ms 设置。</p>
<p>Sender#sendProducerData</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">if</span> (!expiredBatches.isEmpty())</span><br><span class="line">    log.trace(<span class="string">&quot;Expired &#123;&#125; batches in accumulator&quot;</span>, expiredBatches.size());</span><br><span class="line"><span class="keyword">for</span> (ProducerBatch expiredBatch : expiredBatches) &#123;</span><br><span class="line">    String errorMessage = <span class="string">&quot;Expiring &quot;</span> + expiredBatch.recordCount + <span class="string">&quot; record(s) for &quot;</span> + expiredBatch.topicPartition</span><br><span class="line">                + <span class="string">&quot;:&quot;</span> + (now - expiredBatch.createdMs) + <span class="string">&quot; ms has passed since batch creation&quot;</span>;</span><br><span class="line">    failBatch(expiredBatch, -<span class="number">1</span>, NO_TIMESTAMP, <span class="keyword">new</span> TimeoutException(errorMessage), <span class="keyword">false</span>);</span><br><span class="line">    <span class="keyword">if</span> (transactionManager != <span class="keyword">null</span> &amp;&amp; expiredBatch.inRetry()) &#123;</span><br><span class="line">        <span class="comment">// This ensures that no new batches are drained until the current in flight batches are fully resolved.</span></span><br><span class="line">        transactionManager.markSequenceUnresolved(expiredBatch.topicPartition);</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<p>Step7：处理已超时的消息批次，通知该批消息发送失败，即通过设置  KafkaProducer#send 方法返回的凭证中的 FutureRecordMetadata 中的 ProduceRequestResult result，使之调用其 get 方法不会阻塞。</p>
<p>Sender#sendProducerData</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">sensors.updateProduceRequestMetrics(batches);</span><br></pre></td></tr></table></figure>
<p>Step8：收集统计指标，本文不打算详细分析，但后续会专门对 Kafka 的 Metrics 设计进行一个深入的探讨与学习。</p>
<p>Sender#sendProducerData</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">long</span> pollTimeout = Math.min(result.nextReadyCheckDelayMs, notReadyTimeout);</span><br><span class="line">pollTimeout = Math.min(pollTimeout, <span class="keyword">this</span>.accumulator.nextExpiryTimeMs() - now);</span><br><span class="line">pollTimeout = Math.max(pollTimeout, <span class="number">0</span>);</span><br><span class="line"><span class="keyword">if</span> (!result.readyNodes.isEmpty()) &#123;</span><br><span class="line">    log.trace(<span class="string">&quot;Nodes with data ready to send: &#123;&#125;&quot;</span>, result.readyNodes);</span><br><span class="line">    pollTimeout = <span class="number">0</span>;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<p>Step9：设置下一次的发送延时，待补充详细分析。</p>
<p>Sender#sendProducerData</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br></pre></td><td class="code"><pre><span class="line">sendProduceRequests(batches, now);</span><br><span class="line"><span class="function"><span class="keyword">private</span> <span class="keyword">void</span> <span class="title">sendProduceRequests</span><span class="params">(Map&lt;Integer, List&lt;ProducerBatch&gt;&gt; collated, <span class="keyword">long</span> now)</span> </span>&#123;</span><br><span class="line">    <span class="keyword">for</span> (Map.Entry&lt;Integer, List&lt;ProducerBatch&gt;&gt; entry : collated.entrySet())</span><br><span class="line">        sendProduceRequest(now, entry.getKey(), acks, requestTimeoutMs, entry.getValue());</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<p>Step10：该步骤按照 brokerId 分别构建发送请求，即每一个 broker 会将多个  ProducerBatch 一起封装成一个请求进行发送，同一时间，每一个 与 broker 连接只会只能发送一个请求，注意，这里只是构建请求，并最终会通过 NetworkClient#send 方法，将该批数据设置到 NetworkClient 的待发送数据中，此时并没有触发真正的网络调用。</p>
<p>sendProducerData 方法就介绍到这里了，既然这里还没有进行真正的网络请求，那在什么时候触发呢？</p>
<p>我们继续回到 runOnce 方法。</p>
<h6 id="1-2-1-2-NetworkClient-的-poll-方法"><a href="#1-2-1-2-NetworkClient-的-poll-方法" class="headerlink" title="1.2.1.2 NetworkClient 的 poll 方法"></a>1.2.1.2 NetworkClient 的 poll 方法</h6><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br></pre></td><td class="code"><pre><span class="line"> <span class="function"><span class="keyword">public</span> List&lt;ClientResponse&gt; <span class="title">poll</span><span class="params">(<span class="keyword">long</span> timeout, <span class="keyword">long</span> now)</span> </span>&#123;</span><br><span class="line">    ensureActive();</span><br><span class="line"></span><br><span class="line">    <span class="keyword">if</span> (!abortedSends.isEmpty()) &#123;</span><br><span class="line">        <span class="comment">// If there are aborted sends because of unsupported version exceptions or disconnects,</span></span><br><span class="line">        <span class="comment">// handle them immediately without waiting for Selector#poll.</span></span><br><span class="line">        List&lt;ClientResponse&gt; responses = <span class="keyword">new</span> ArrayList&lt;&gt;();</span><br><span class="line">        handleAbortedSends(responses);</span><br><span class="line">        completeResponses(responses);</span><br><span class="line">        <span class="keyword">return</span> responses;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="keyword">long</span> metadataTimeout = metadataUpdater.maybeUpdate(now);   <span class="comment">// @1</span></span><br><span class="line">    <span class="keyword">try</span> &#123;</span><br><span class="line">        <span class="keyword">this</span>.selector.poll(Utils.min(timeout, metadataTimeout, defaultRequestTimeoutMs));    <span class="comment">// @2</span></span><br><span class="line">    &#125; <span class="keyword">catch</span> (IOException e) &#123;</span><br><span class="line">        log.error(<span class="string">&quot;Unexpected error during I/O&quot;</span>, e);</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="comment">// process completed actions</span></span><br><span class="line">    <span class="keyword">long</span> updatedNow = <span class="keyword">this</span>.time.milliseconds();</span><br><span class="line">    List&lt;ClientResponse&gt; responses = <span class="keyword">new</span> ArrayList&lt;&gt;();            <span class="comment">// @3</span></span><br><span class="line">    handleCompletedSends(responses, updatedNow);</span><br><span class="line">    handleCompletedReceives(responses, updatedNow);</span><br><span class="line">    handleDisconnections(responses, updatedNow);</span><br><span class="line">    handleConnections();</span><br><span class="line">    handleInitiateApiVersionRequests(updatedNow);</span><br><span class="line">    handleTimedOutRequests(responses, updatedNow);</span><br><span class="line">    completeResponses(responses);                                               <span class="comment">// @4</span></span><br><span class="line">    <span class="keyword">return</span> responses;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<p>本文并不会详细深入探讨其网络实现部分，Kafka 的 网络通讯后续我会专门详细的介绍，在这里先点出其关键点。<br>代码@1：尝试更新云数据。<br>代码@2：触发真正的网络通讯，该方法中会通过收到调用 NIO 中的 Selector#select() 方法，对通道的读写就绪事件进行处理，当写事件就绪后，就会将通道中的消息发送到远端的 broker。<br>代码@3：然后会消息发送，消息接收、断开连接、API版本，超时等结果进行收集。<br>代码@4：并依次对结果进行唤醒，此时会将响应结果设置到  KafkaProducer#send 方法返回的凭证中，从而唤醒发送客户端，完成一次完整的消息发送流程。</p>
<p>Sender 发送线程的流程就介绍到这里了，接下来首先给出一张流程图，然后对上述流程中一些关键的方法再补充深入探讨一下。</p>
<h5 id="1-2-2-run-方法流程图"><a href="#1-2-2-run-方法流程图" class="headerlink" title="1.2.2 run 方法流程图"></a>1.2.2 run 方法流程图</h5><p><img src="https://img-blog.csdnimg.cn/20191117182748452.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ByZXN0aWdlZGluZw==,size_16,color_FFFFFF,t_70" alt="在这里插入图片描述"><br>根据上面的源码分析得出上述流程图，图中对重点步骤也详细标注了其关键点。下面我们对上述流程图中 Sender 线程依赖的相关类的核心方法进行解读，以便加深 Sender 线程的理解。</p>
<p>由于在讲解 Sender 发送流程中，大部分都是调用 RecordAccumulator 方法来实现其特定逻辑，故接下来重点对上述涉及到RecordAccumulator 的方法进行一个详细剖析，加强对 Sender 流程的理解。</p>
<h2 id="2、RecordAccumulator-核心方法详解"><a href="#2、RecordAccumulator-核心方法详解" class="headerlink" title="2、RecordAccumulator 核心方法详解"></a>2、RecordAccumulator 核心方法详解</h2><h4 id="2-1-RecordAccumulator-的-ready-方法详解"><a href="#2-1-RecordAccumulator-的-ready-方法详解" class="headerlink" title="2.1 RecordAccumulator 的 ready 方法详解"></a>2.1 RecordAccumulator 的 ready 方法详解</h4><p>该方法主要就是根据缓存区中的消息，判断哪些分区已经达到发送条件。</p>
<p>RecordAccumulator#ready</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br></pre></td><td class="code"><pre><span class="line"><span class="function"><span class="keyword">public</span> ReadyCheckResult <span class="title">ready</span><span class="params">(Cluster cluster, <span class="keyword">long</span> nowMs)</span> </span>&#123;</span><br><span class="line">    Set&lt;Node&gt; readyNodes = <span class="keyword">new</span> HashSet&lt;&gt;();</span><br><span class="line">    <span class="keyword">long</span> nextReadyCheckDelayMs = Long.MAX_VALUE;</span><br><span class="line">    Set&lt;String&gt; unknownLeaderTopics = <span class="keyword">new</span> HashSet&lt;&gt;();</span><br><span class="line"></span><br><span class="line">    <span class="keyword">boolean</span> exhausted = <span class="keyword">this</span>.free.queued() &gt; <span class="number">0</span>;</span><br><span class="line">    <span class="keyword">for</span> (Map.Entry&lt;TopicPartition, Deque&lt;ProducerBatch&gt;&gt; entry : <span class="keyword">this</span>.batches.entrySet()) &#123;   <span class="comment">// @1</span></span><br><span class="line">        TopicPartition part = entry.getKey();</span><br><span class="line">        Deque&lt;ProducerBatch&gt; deque = entry.getValue();</span><br><span class="line"></span><br><span class="line">        Node leader = cluster.leaderFor(part);   <span class="comment">// @2</span></span><br><span class="line">        <span class="keyword">synchronized</span> (deque) &#123;</span><br><span class="line">            <span class="keyword">if</span> (leader == <span class="keyword">null</span> &amp;&amp; !deque.isEmpty()) &#123;   <span class="comment">// @3</span></span><br><span class="line">                <span class="comment">// This is a partition for which leader is not known, but messages are available to send.</span></span><br><span class="line">                <span class="comment">// Note that entries are currently not removed from batches when deque is empty.</span></span><br><span class="line">                unknownLeaderTopics.add(part.topic());</span><br><span class="line">            &#125; <span class="keyword">else</span> <span class="keyword">if</span> (!readyNodes.contains(leader) &amp;&amp; !isMuted(part, nowMs)) &#123;    <span class="comment">// @4</span></span><br><span class="line">                ProducerBatch batch = deque.peekFirst();</span><br><span class="line">                <span class="keyword">if</span> (batch != <span class="keyword">null</span>) &#123;</span><br><span class="line">                    <span class="keyword">long</span> waitedTimeMs = batch.waitedTimeMs(nowMs);</span><br><span class="line">                    <span class="keyword">boolean</span> backingOff = batch.attempts() &gt; <span class="number">0</span> &amp;&amp; waitedTimeMs &lt; retryBackoffMs;</span><br><span class="line">                    <span class="keyword">long</span> timeToWaitMs = backingOff ? retryBackoffMs : lingerMs;</span><br><span class="line">                    <span class="keyword">boolean</span> full = deque.size() &gt; <span class="number">1</span> || batch.isFull();</span><br><span class="line">                    <span class="keyword">boolean</span> expired = waitedTimeMs &gt;= timeToWaitMs;</span><br><span class="line">                    <span class="keyword">boolean</span> sendable = full || expired || exhausted || closed || flushInProgress();</span><br><span class="line">                    <span class="keyword">if</span> (sendable &amp;&amp; !backingOff) &#123;   <span class="comment">// @5</span></span><br><span class="line">                        readyNodes.add(leader);</span><br><span class="line">                    &#125; <span class="keyword">else</span> &#123;</span><br><span class="line">                        <span class="keyword">long</span> timeLeftMs = Math.max(timeToWaitMs - waitedTimeMs, <span class="number">0</span>);</span><br><span class="line">                        <span class="comment">// Note that this results in a conservative estimate since an un-sendable partition may have</span></span><br><span class="line">                        <span class="comment">// a leader that will later be found to have sendable data. However, this is good enough</span></span><br><span class="line">                        <span class="comment">// since we&#x27;ll just wake up and then sleep again for the remaining time.</span></span><br><span class="line">                        nextReadyCheckDelayMs = Math.min(timeLeftMs, nextReadyCheckDelayMs);   </span><br><span class="line">                    &#125;</span><br><span class="line">                &#125;</span><br><span class="line">            &#125;</span><br><span class="line">        &#125;</span><br><span class="line">    &#125;</span><br><span class="line">    <span class="keyword">return</span> <span class="keyword">new</span> ReadyCheckResult(readyNodes, nextReadyCheckDelayMs, unknownLeaderTopics);</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<p>代码@1：对生产者缓存区 ConcurrentHashMap&lt;TopicPartition, Deque&lt; ProducerBatch&gt;&gt; batches 遍历，从中挑选已准备好的消息批次。<br>代码@2：从生产者元数据缓存中尝试查找分区(TopicPartition) 的 leader 信息，如果不存在，当将该 topic 添加到 unknownLeaderTopics (代码@3)，稍后会发送元数据更新请求去 broker 端查找分区的路由信息。<br>代码@4：如果不在 readyNodes 中就需要判断是否满足条件，isMuted 与顺序消息有关，本文暂时不关注，在后面的顺序消息部分会重点探讨。<br>代码@5：这里就是判断是否准备好的条件，先一个一个来解读局部变量的含义。</p>
<ul>
<li>long waitedTimeMs<br>该 ProducerBatch 已等待的时长，等于当前时间戳 与 ProducerBatch 的 lastAttemptMs 之差，在 ProducerBatch 创建时或需要重试时会将当前的时间赋值给lastAttemptMs。</li>
<li>retryBackoffMs<br>当发生异常时发起重试之前的等待时间，默认为 100ms，可通过属性 retry.backoff.ms 配置。</li>
<li>batch.attempts()<br>该批次当前已重试的次数。</li>
<li>backingOff<br>后台发送是否关闭，即如果需要重试并且等待时间小于 retryBackoffMs ，则 backingOff = true，也意味着该批次未准备好。</li>
<li>timeToWaitMs<br>send 线程发送消息需要的等待时间，如果 backingOff  为 true，表示该批次是在重试，并且等待时间小于系统设置的需要等待时间，这种情况下 timeToWaitMs = retryBackoffMs 。否则需要等待的时间为 lingerMs。</li>
<li>boolean full<br>该批次是否已满，如果两个条件中的任意一个满足即为 true。<ul>
<li>Deque&lt; ProducerBatch&gt; 该队列的个数大于1，表示肯定有一个 ProducerBatch 已写满。 </li>
<li>ProducerBatch 已写满。</li>
</ul>
</li>
<li>boolean expired<br>是否过期，等于已经等待的时间是否大于需要等待的时间，如果把发送看成定时发送的话，expired 为 true 表示定时器已到达触发点，即需要执行。</li>
<li>boolean exhausted<br>当前生产者缓存已不够，创建新的 ProducerBatch 时阻塞在申请缓存空间的线程大于0，此时应立即将缓存区中的消息立即发送到服务器。</li>
<li>boolean sendable<br>是否可发送。其满足下面的任意一个条件即可：<ul>
<li>   该批次已写满。(full = true)。</li>
<li>   已等待系统规定的时长。（expired = true）</li>
<li>   发送者内部缓存区已耗尽并且有新的线程需要申请(exhausted = true)。</li>
<li>   该发送者的 close 方法被调用(close = true)。</li>
<li>   该发送者的 flush 方法被调用。</li>
</ul>
</li>
</ul>
<h4 id="2-2-RecordAccumulator-的-drain方法详解"><a href="#2-2-RecordAccumulator-的-drain方法详解" class="headerlink" title="2.2 RecordAccumulator 的 drain方法详解"></a>2.2 RecordAccumulator 的 drain方法详解</h4><p>RecordAccumulator#drain</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">public</span> Map&lt;Integer, List&lt;ProducerBatch&gt;&gt; drain(Cluster cluster, Set&lt;Node&gt; nodes, <span class="keyword">int</span> maxSize, <span class="keyword">long</span> now) &#123; <span class="comment">// @1</span></span><br><span class="line">    <span class="keyword">if</span> (nodes.isEmpty())</span><br><span class="line">        <span class="keyword">return</span> Collections.emptyMap();</span><br><span class="line"></span><br><span class="line">    Map&lt;Integer, List&lt;ProducerBatch&gt;&gt; batches = <span class="keyword">new</span> HashMap&lt;&gt;();</span><br><span class="line">    <span class="keyword">for</span> (Node node : nodes) &#123;                                                                                                                              </span><br><span class="line">        List&lt;ProducerBatch&gt; ready = drainBatchesForOneNode(cluster, node, maxSize, now);                      <span class="comment">// @2</span></span><br><span class="line">        batches.put(node.id(), ready);</span><br><span class="line">    &#125;</span><br><span class="line">    <span class="keyword">return</span> batches;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<p>代码@1：我们首先来介绍该方法的参数：</p>
<ul>
<li>Cluster cluster<br>集群信息。</li>
<li>Set&lt; Node&gt; nodes<br>已准备好的节点集合。</li>
<li>int maxSize<br>一次请求最大的字节数。</li>
<li>long now<br>当前时间。</li>
</ul>
<p>代码@2：遍历所有节点，调用 drainBatchesForOneNode 方法抽取数据，组装成 Map&lt;Integer /** brokerId */, List&lt; ProducerBatch&gt;&gt; batches。</p>
<p>接下来重点来看一下 drainBatchesForOneNode。<br>RecordAccumulator#drainBatchesForOneNode</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br></pre></td><td class="code"><pre><span class="line"><span class="function"><span class="keyword">private</span> List&lt;ProducerBatch&gt; <span class="title">drainBatchesForOneNode</span><span class="params">(Cluster cluster, Node node, <span class="keyword">int</span> maxSize, <span class="keyword">long</span> now)</span> </span>&#123;</span><br><span class="line">    <span class="keyword">int</span> size = <span class="number">0</span>;</span><br><span class="line">    List&lt;PartitionInfo&gt; parts = cluster.partitionsForNode(node.id());   <span class="comment">// @1</span></span><br><span class="line">    List&lt;ProducerBatch&gt; ready = <span class="keyword">new</span> ArrayList&lt;&gt;();</span><br><span class="line">    <span class="keyword">int</span> start = drainIndex = drainIndex % parts.size();                        <span class="comment">// @2</span></span><br><span class="line">    <span class="keyword">do</span> &#123;                                                                                                <span class="comment">// @3 </span></span><br><span class="line">        PartitionInfo part = parts.get(drainIndex);</span><br><span class="line">        TopicPartition tp = <span class="keyword">new</span> TopicPartition(part.topic(), part.partition()); </span><br><span class="line">        <span class="keyword">this</span>.drainIndex = (<span class="keyword">this</span>.drainIndex + <span class="number">1</span>) % parts.size();                     </span><br><span class="line">            </span><br><span class="line">        <span class="keyword">if</span> (isMuted(tp, now))</span><br><span class="line">            <span class="keyword">continue</span>;</span><br><span class="line"></span><br><span class="line">        Deque&lt;ProducerBatch&gt; deque = getDeque(tp);                              <span class="comment">// @4</span></span><br><span class="line">        <span class="keyword">if</span> (deque == <span class="keyword">null</span>)</span><br><span class="line">            <span class="keyword">continue</span>;</span><br><span class="line"></span><br><span class="line">        <span class="keyword">synchronized</span> (deque) &#123;</span><br><span class="line">            <span class="comment">// invariant: !isMuted(tp,now) &amp;&amp; deque != null</span></span><br><span class="line">            ProducerBatch first = deque.peekFirst();                                         <span class="comment">// @5</span></span><br><span class="line">            <span class="keyword">if</span> (first == <span class="keyword">null</span>)</span><br><span class="line">                <span class="keyword">continue</span>;</span><br><span class="line"></span><br><span class="line">            <span class="comment">// first != null</span></span><br><span class="line">            <span class="keyword">boolean</span> backoff = first.attempts() &gt; <span class="number">0</span> &amp;&amp; first.waitedTimeMs(now) &lt; retryBackoffMs;   <span class="comment">// @6</span></span><br><span class="line">            <span class="comment">// Only drain the batch if it is not during backoff period.</span></span><br><span class="line">            <span class="keyword">if</span> (backoff)                                                                                     </span><br><span class="line">                <span class="keyword">continue</span>;</span><br><span class="line"></span><br><span class="line">            <span class="keyword">if</span> (size + first.estimatedSizeInBytes() &gt; maxSize &amp;&amp; !ready.isEmpty()) &#123;     <span class="comment">// @7</span></span><br><span class="line">                <span class="keyword">break</span>;</span><br><span class="line">            &#125; <span class="keyword">else</span> &#123;</span><br><span class="line">                <span class="keyword">if</span> (shouldStopDrainBatchesForPartition(first, tp))                                  </span><br><span class="line">                    <span class="keyword">break</span>;</span><br><span class="line"></span><br><span class="line">                <span class="comment">// 这里省略与事务消息相关的代码，后续会重点学习。</span></span><br><span class="line">                batch.close();                                                                                            <span class="comment">// @8</span></span><br><span class="line">                size += batch.records().sizeInBytes();</span><br><span class="line">                ready.add(batch);                                                                            </span><br><span class="line"></span><br><span class="line">                batch.drained(now);                                                                             </span><br><span class="line">            &#125;</span><br><span class="line">        &#125;</span><br><span class="line">    &#125; <span class="keyword">while</span> (start != drainIndex);</span><br><span class="line">    <span class="keyword">return</span> ready;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<p>代码@1：根据 brokerId 获取该 broker 上的所有主分区。<br>代码@2：初始化 start。这里首先来阐述一下 start 与 drainIndex 。</p>
<ul>
<li>start 当前开始遍历的分区序号。</li>
<li>drainIndex 上次抽取的队列索引后，这里主要是为了每个队列都是从零号分区开始抽取。</li>
</ul>
<p>代码@3：循环从缓存区抽取对应分区中累积的数据。<br>代码@4：根据 topic + 分区号从生产者发送缓存区中获取已累积的双端Queue。<br>代码@5：从双端队列的头部获取一个元素。（消息追加时是追加到队列尾部）。<br>代码@6：如果当前批次是重试，并且还未到阻塞时间，则跳过该分区。<br>代码@7：如果当前已抽取的消息总大小 加上新的消息已超过 maxRequestSize，则结束抽取。<br>代码@8：将当前批次加入到已准备集合中，并关闭该批次，即不在允许向该批次中追加消息。</p>
<p>关于消息发送就介绍到这里，NetworkClient 的 poll 方法内部会调用 Selector 执行就绪事件的选择，并将抽取的消息通过网络发送到 Broker 服务器，关于网络后面的具体实现，将在后续文章中单独介绍。</p>
</div>

			<script src="https://my.openwrite.cn/js/readmore.js" type="text/javascript"></script>
			<script>
			var isMobile = navigator.userAgent.match(/(phone|pad|pod|iPhone|iPod|ios|iPad|Android|Mobile|BlackBerry|IEMobile|MQQBrowser|JUC|Fennec|wOSBrowser|BrowserNG|WebOS|Symbian|Windows Phone)/i);
			if (!isMobile) {
			    var btw = new BTWPlugin();
			    btw.init({
			        "id": "vip-container",
			        "blogId": "18019-1573088808868-542",
			        "name": "中间件兴趣圈",
			        "qrcode": "https://img-blog.csdnimg.cn/20190314214003962.jpg",
			        "keyword": "more"
			    });
			}
			</script>
		
      
    </div>
    
    
    

    

    

    

    <footer class="post-footer">
      

      
      
      

      
        <div class="post-nav">
          <div class="post-nav-next post-nav-item">
            
              <a href="/posts/a79560fc.html" rel="next" title="Netty4 Channel 概述">
                <i class="fa fa-chevron-left"></i> Netty4 Channel 概述
              </a>
            
          </div>

          <span class="post-nav-divider"></span>

          <div class="post-nav-prev post-nav-item">
            
              <a href="/posts/a618bda.html" rel="prev" title="源码分析 Alibaba sentinel 滑动窗口实现原理(文末附原理图)">
                源码分析 Alibaba sentinel 滑动窗口实现原理(文末附原理图) <i class="fa fa-chevron-right"></i>
              </a>
            
          </div>
        </div>
      

      
      
    </footer>
  </div>
  
  
  
  </article>



    <div class="post-spread">
      
    </div>
  </div>


          </div>
          


          

  



        </div>
        
          
  
  <div class="sidebar-toggle">
    <div class="sidebar-toggle-line-wrap">
      <span class="sidebar-toggle-line sidebar-toggle-line-first"></span>
      <span class="sidebar-toggle-line sidebar-toggle-line-middle"></span>
      <span class="sidebar-toggle-line sidebar-toggle-line-last"></span>
    </div>
  </div>

  <aside id="sidebar" class="sidebar">
    
    <div class="sidebar-inner">

      

      
        <ul class="sidebar-nav motion-element">
          <li class="sidebar-nav-toc sidebar-nav-active" data-target="post-toc-wrap">
            文章目录
          </li>
          <li class="sidebar-nav-overview" data-target="site-overview-wrap">
            站点概览
          </li>
        </ul>
      

      <section class="site-overview-wrap sidebar-panel">
        <div class="site-overview">
          <div class="site-author motion-element" itemprop="author" itemscope itemtype="http://schema.org/Person">
            
              <p class="site-author-name" itemprop="name"></p>
              <p class="site-description motion-element" itemprop="description"></p>
          </div>

          <nav class="site-state motion-element">

            
              <div class="site-state-item site-state-posts">
              
                <a href="/archives/">
              
                  <span class="site-state-item-count">139</span>
                  <span class="site-state-item-name">日志</span>
                </a>
              </div>
            

            
              
              
              <div class="site-state-item site-state-categories">
                <a href="/categories/index.html">
                  <span class="site-state-item-count">18</span>
                  <span class="site-state-item-name">分类</span>
                </a>
              </div>
            

            

          </nav>

          

          

          
          

          
          

          

        </div>
      </section>

      
      <!--noindex-->
        <section class="post-toc-wrap motion-element sidebar-panel sidebar-panel-active">
          <div class="post-toc">

            
              
            

            
              <div class="post-toc-content"><ol class="nav"><li class="nav-item nav-level-2"><a class="nav-link" href="#1%E3%80%81Sender-%E7%BA%BF%E7%A8%8B%E8%AF%A6%E8%A7%A3"><span class="nav-number">1.</span> <span class="nav-text">1、Sender 线程详解</span></a><ol class="nav-child"><li class="nav-item nav-level-4"><a class="nav-link" href="#1-1-%E7%B1%BB%E5%9B%BE"><span class="nav-number">1.0.1.</span> <span class="nav-text">1.1 类图</span></a></li><li class="nav-item nav-level-4"><a class="nav-link" href="#1-2-run-%E6%96%B9%E6%B3%95%E8%AF%A6%E8%A7%A3"><span class="nav-number">1.0.2.</span> <span class="nav-text">1.2 run 方法详解</span></a><ol class="nav-child"><li class="nav-item nav-level-5"><a class="nav-link" href="#1-2-1-runOnce-%E8%AF%A6%E8%A7%A3"><span class="nav-number">1.0.2.1.</span> <span class="nav-text">1.2.1 runOnce 详解</span></a><ol class="nav-child"><li class="nav-item nav-level-6"><a class="nav-link" href="#1-1-2-1-sendProducerData"><span class="nav-number">1.0.2.1.1.</span> <span class="nav-text">1.1.2.1 sendProducerData</span></a></li><li class="nav-item nav-level-6"><a class="nav-link" href="#1-2-1-2-NetworkClient-%E7%9A%84-poll-%E6%96%B9%E6%B3%95"><span class="nav-number">1.0.2.1.2.</span> <span class="nav-text">1.2.1.2 NetworkClient 的 poll 方法</span></a></li></ol></li><li class="nav-item nav-level-5"><a class="nav-link" href="#1-2-2-run-%E6%96%B9%E6%B3%95%E6%B5%81%E7%A8%8B%E5%9B%BE"><span class="nav-number">1.0.2.2.</span> <span class="nav-text">1.2.2 run 方法流程图</span></a></li></ol></li></ol></li></ol></li><li class="nav-item nav-level-2"><a class="nav-link" href="#2%E3%80%81RecordAccumulator-%E6%A0%B8%E5%BF%83%E6%96%B9%E6%B3%95%E8%AF%A6%E8%A7%A3"><span class="nav-number">2.</span> <span class="nav-text">2、RecordAccumulator 核心方法详解</span></a><ol class="nav-child"><li class="nav-item nav-level-4"><a class="nav-link" href="#2-1-RecordAccumulator-%E7%9A%84-ready-%E6%96%B9%E6%B3%95%E8%AF%A6%E8%A7%A3"><span class="nav-number">2.0.1.</span> <span class="nav-text">2.1 RecordAccumulator 的 ready 方法详解</span></a></li><li class="nav-item nav-level-4"><a class="nav-link" href="#2-2-RecordAccumulator-%E7%9A%84-drain%E6%96%B9%E6%B3%95%E8%AF%A6%E8%A7%A3"><span class="nav-number">2.0.2.</span> <span class="nav-text">2.2 RecordAccumulator 的 drain方法详解</span></a></li></ol></li></ol></li></ol></div>
            

          </div>
        </section>
      <!--/noindex-->
      

      

    </div>
  </aside>


        
      </div>
    </main>

    <footer id="footer" class="footer">
      <div class="footer-inner">
        <div class="copyright">&copy; <span itemprop="copyrightYear">2021</span>
  <span class="with-love">
    <i class="fa fa-user"></i>
  </span>
  <span class="author" itemprop="copyrightHolder">中间件兴趣圈</span>

  
</div>


  <div class="powered-by">由 <a class="theme-link" target="_blank" href="https://hexo.io">Hexo</a> 强力驱动</div>



  <span class="post-meta-divider">|</span>



  <div class="theme-info">主题 &mdash; <a class="theme-link" target="_blank" href="https://github.com/iissnan/hexo-theme-next">NexT.Muse</a> v5.1.4</div>




        
<div class="busuanzi-count">
  <script async src="https://dn-lbstatics.qbox.me/busuanzi/2.3/busuanzi.pure.mini.js"></script>

  
    <span class="site-uv">
      <i class="fa fa-user"></i>
      <span class="busuanzi-value" id="busuanzi_value_site_uv"></span>
      
    </span>
  

  
    <span class="site-pv">
      <i class="fa fa-eye"></i>
      <span class="busuanzi-value" id="busuanzi_value_site_pv"></span>
      
    </span>
  
</div>








        
      </div>
    </footer>

    
      <div class="back-to-top">
        <i class="fa fa-arrow-up"></i>
        
      </div>
    

    

  </div>

  

<script type="text/javascript">
  if (Object.prototype.toString.call(window.Promise) !== '[object Function]') {
    window.Promise = null;
  }
</script>









  












  
  
    <script type="text/javascript" src="/lib/jquery/index.js?v=2.1.3"></script>
  

  
  
    <script type="text/javascript" src="/lib/fastclick/lib/fastclick.min.js?v=1.0.6"></script>
  

  
  
    <script type="text/javascript" src="/lib/jquery_lazyload/jquery.lazyload.js?v=1.9.7"></script>
  

  
  
    <script type="text/javascript" src="/lib/velocity/velocity.min.js?v=1.2.1"></script>
  

  
  
    <script type="text/javascript" src="/lib/velocity/velocity.ui.min.js?v=1.2.1"></script>
  

  
  
    <script type="text/javascript" src="/lib/fancybox/source/jquery.fancybox.pack.js?v=2.1.5"></script>
  


  


  <script type="text/javascript" src="/js/src/utils.js?v=5.1.4"></script>

  <script type="text/javascript" src="/js/src/motion.js?v=5.1.4"></script>



  
  

  
  <script type="text/javascript" src="/js/src/scrollspy.js?v=5.1.4"></script>
<script type="text/javascript" src="/js/src/post-details.js?v=5.1.4"></script>



  


  <script type="text/javascript" src="/js/src/bootstrap.js?v=5.1.4"></script>



  


  




	





  





  












  





  

  
  <script src="https://cdn1.lncld.net/static/js/av-core-mini-0.6.4.js"></script>
  <script>AV.initialize("NNEhOL0iOcflg8f1U3HUqiCq-gzGzoHsz", "7kSmkbbb3DktmHALlShDsBUF");</script>
  <script>
    function showTime(Counter) {
      var query = new AV.Query(Counter);
      var entries = [];
      var $visitors = $(".leancloud_visitors");

      $visitors.each(function () {
        entries.push( $(this).attr("id").trim() );
      });

      query.containedIn('url', entries);
      query.find()
        .done(function (results) {
          var COUNT_CONTAINER_REF = '.leancloud-visitors-count';

          if (results.length === 0) {
            $visitors.find(COUNT_CONTAINER_REF).text(0);
            return;
          }

          for (var i = 0; i < results.length; i++) {
            var item = results[i];
            var url = item.get('url');
            var time = item.get('time');
            var element = document.getElementById(url);

            $(element).find(COUNT_CONTAINER_REF).text(time);
          }
          for(var i = 0; i < entries.length; i++) {
            var url = entries[i];
            var element = document.getElementById(url);
            var countSpan = $(element).find(COUNT_CONTAINER_REF);
            if( countSpan.text() == '') {
              countSpan.text(0);
            }
          }
        })
        .fail(function (object, error) {
          console.log("Error: " + error.code + " " + error.message);
        });
    }

    function addCount(Counter) {
      var $visitors = $(".leancloud_visitors");
      var url = $visitors.attr('id').trim();
      var title = $visitors.attr('data-flag-title').trim();
      var query = new AV.Query(Counter);

      query.equalTo("url", url);
      query.find({
        success: function(results) {
          if (results.length > 0) {
            var counter = results[0];
            counter.fetchWhenSave(true);
            counter.increment("time");
            counter.save(null, {
              success: function(counter) {
                var $element = $(document.getElementById(url));
                $element.find('.leancloud-visitors-count').text(counter.get('time'));
              },
              error: function(counter, error) {
                console.log('Failed to save Visitor num, with error message: ' + error.message);
              }
            });
          } else {
            var newcounter = new Counter();
            /* Set ACL */
            var acl = new AV.ACL();
            acl.setPublicReadAccess(true);
            acl.setPublicWriteAccess(true);
            newcounter.setACL(acl);
            /* End Set ACL */
            newcounter.set("title", title);
            newcounter.set("url", url);
            newcounter.set("time", 1);
            newcounter.save(null, {
              success: function(newcounter) {
                var $element = $(document.getElementById(url));
                $element.find('.leancloud-visitors-count').text(newcounter.get('time'));
              },
              error: function(newcounter, error) {
                console.log('Failed to create');
              }
            });
          }
        },
        error: function(error) {
          console.log('Error:' + error.code + " " + error.message);
        }
      });
    }

    $(function() {
      var Counter = AV.Object.extend("Counter");
      if ($('.leancloud_visitors').length == 1) {
        addCount(Counter);
      } else if ($('.post-title-link').length > 1) {
        showTime(Counter);
      }
    });
  </script>



  

  

  
  

  

  

  

</body>
</html>
